package com.example.day04.controller.union;

import com.example.day04.controller.bean.WaterSemsor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

public class Demo1 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);
        DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3, 4);
        DataStreamSource<Integer> source2 = env.fromElements(11, 22, 33, 44);
        DataStreamSource<String> source3 = env.fromElements("1", "2", "3", "4");
// 将两个或多个数据流联合来创建一个包含所有流中数据的新流
//        DataStream<Integer> union = source1.union(source2).union(source3.map(Integer::valueOf));

//        union.print("zhao");
        ConnectedStreams<Integer, String> connect = source1.connect(source3);
        // “连接” 两个数据流并保留各自的类型。connect 允许在两个流的处理逻辑之间共享状态
        SingleOutputStreamOperator<Object> map = connect.map(new CoMapFunction<Integer, String, Object>() {
            @Override
            public Object map1(Integer value) throws Exception {
                return value;
            }

            @Override
            public Object map2(String value) throws Exception {
                return value;
            }
        });
        map.print("zhao");

        env.execute();
    }
}
